package com.shujia.flink.core;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo02WordCountOnBatch {
    public static void main(String[] args) throws Exception {

        // 构建环境
        // 默认还是会进行流处理，数据是被一条一条处理的，结果是连续变化的
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 切换运行模式
        /*
         * 可以指定三种模式：
         * BATCH：批处理模式，只能作用在有界流上，计算结果是直接输出最终的结果
         * STREAMING：默认的运行方式，流处理模式，即可以作用在无界流上也可以作用在有界流上，结果是连续变化的
         * AUTOMATIC：自动识别，所有source都是有界流时，则为BATCH模式，只要有一个source是无界流就会设为STREAMING流处理模式
         */
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        // 构建DataStream
        DataStream<String> fileDS = env.readTextFile("flink/data/words.txt"); // 从文件构建的流是有界流

        fileDS
                .flatMap((line, out) -> {
                    for (String word : line.split(",")) {
                        out.collect(word);
                    }
                }, Types.STRING)
                .map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(kv -> kv.f0, Types.STRING)
                .sum(1)
                .print();


        env.execute();

    }
}
